package com.cayden.components.kafka;

import com.cayden.config.boot.Apps;
import lombok.extern.slf4j.Slf4j;
import org.springframework.core.env.Environment;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.URI;

/**
 * @author cayden
 * @date 2021-10-25 15:23
 */
@Slf4j
public class KafkaHasConnectionUtils {

    public static boolean hasConnection(){
        Environment environment = Apps.getEnvironment();
        String ipPort = environment.getProperty("spring.kafka.producer.bootstrap-servers");

        System.out.println(ipPort);
        URI uri = URI.create("http://" + ipPort);

        String host = uri.getHost();

        int port = uri.getPort();

        return isHostConnectable(host, port);
    }

    /**
     * 判断kafka服务器，能否正常连接
     *
     * @param host
     * @param port
     * @return
     */
    static boolean isHostConnectable(String host, int port) {
        Socket socket = new Socket();
        try {
            socket.connect(new InetSocketAddress(host, port), 2000);
        } catch (IOException e) {
            log.error("kafka 连接失败,请检查连接");
            return false;
        } finally {
            try {
                socket.close();
            } catch (IOException e) {
                log.error("socket 关闭失败",e);
            }
        }
        return true;
    }
}
